package com.yupi.springbootinit.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import com.yupi.springbootinit.utils.RabbitMqUtils;

import java.nio.charset.StandardCharsets;
import java.util.HashSet;
import java.util.Set;

public class MultiConsumer {

    private static final String TASK_QUEUE_NAME = "multi_queue";
    private static final Set<Long> acknowledgedMessages = new HashSet<>();

    public static void main(String[] argv) throws Exception {
        // 建立连接
        for (int i = 0; i < 2; i++) {
            final Channel channel = RabbitMqUtils.getChannel();

            channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

            channel.basicQos(1);

            // 定义了如何处理消息
            int finalI = i;
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                // 在 RabbitMQ 中，每个传递给消费者的消息都有一个唯一的交付标签，用于标识该消息。
                // 交付标签是一个64位的长整型数字，在消息传递过程中用于确保消息的有序处理和确认。
                long deliveryTag = delivery.getEnvelope().getDeliveryTag();
                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);

                // 当一个消费者确认了消息后，另一个消费者就不应再次确认相同的消息。
                // 为了达到这个目标，我们可以在每个消费者内部维护一个已确认消息的集合，并在处理消息前检查消息是否已经确认过。
                // 如果消息已经确认过了，就不再进行确认。下面是修改后的代码：
                if (!acknowledgedMessages.contains(deliveryTag)) {
                    try {
                        // 处理工作
                        System.out.println(" [x] Received '" + "编号:" + finalI + ":" + message + "'");
                        // 停 20 秒，模拟机器处理能力有限
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        // ↓ 用于拒绝（Nack）一条消息的确认，并将消息重新放回队列。
                        channel.basicNack(deliveryTag, false, false);
                    } finally {
                        System.out.println(" [x] Done");
                        channel.basicAck(deliveryTag, false);
                        acknowledgedMessages.add(deliveryTag);
                    }
                } else {
                    System.out.println(" [x] Message with deliveryTag " + deliveryTag + " already acknowledged.");
                }
            };
            // 开启消费监听
            channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> {
            });
        }
    }
}